package edu.indiana.d2i.mapreduce;

import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.SortedMap;
import java.util.TreeMap;

import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.ddl.ComparatorType;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import edu.indiana.d2i.util.hector.HectorHadoopClient;

@SuppressWarnings("rawtypes")
public class HectorColumnFamilyRecordReader extends
		RecordReader<Object, List<HColumn<?, byte[]>>> {
	protected Object key = null;
	protected List<HColumn<?, byte[]>> value = null;
	protected HectorColumnFamilySplit split = null;

	protected int totalRows;
	protected int rowsRead;
	
	protected HectorHadoopClient client = null;
	protected SortedMap<?, List<HColumn<?, byte[]>>> results = null;
	
	@Override
	public void close() throws IOException {
		// client.close();
	}

	@Override
	public Object getCurrentKey() throws IOException, InterruptedException {
		return key;
	}

	@Override
	public List<HColumn<?, byte[]>> getCurrentValue() throws IOException,
			InterruptedException {
		return value;
	}

	@Override
	public float getProgress() throws IOException, InterruptedException {
		return (float) rowsRead / (float) totalRows;
	}

	@Override
	@SuppressWarnings("unchecked")
	public void initialize(InputSplit inputSplit,
			TaskAttemptContext taskAttemptContext) throws IOException,
			InterruptedException {
		this.split = (HectorColumnFamilySplit) inputSplit;
		this.totalRows = this.split.getKeyList().size();
		this.rowsRead = 0;

		Configuration conf = taskAttemptContext.getConfiguration();		
		try {
			Class<?> clientClass = HectorConfigHelper.getHectorClientClass(conf);
			Constructor<HectorHadoopClient> constructor = (Constructor<HectorHadoopClient>) clientClass
					.getConstructor(Configuration.class, InputSplit.class);
			client = constructor.newInstance(conf, split);
		} catch (SecurityException e) {
			e.printStackTrace();
		} catch (NoSuchMethodException e) {
			e.printStackTrace();
		} catch (IllegalArgumentException e) {
			e.printStackTrace();
		} catch (InstantiationException e) {
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			e.printStackTrace();
		} catch (InvocationTargetException e) {
			e.printStackTrace();
		}
	}

	@Override
	public boolean nextKeyValue() throws IOException, InterruptedException {
		if (rowsRead >= totalRows) // ??
			return false;

		if (results == null) {
			results = client.read();
			if (results == null)
				return false;
		}
		if (results.size() == 0) {
			results = client.read();
			if (results == null || results.size() == 0) {
				System.err.println("just some partial results returned!!");
				return false;
			}
		}

		key = results.firstKey();
		value = results.get(key);
		results.remove(key); // reference ??
		rowsRead++;
		return true;
	}
}